package demo;

import java.util.Arrays;

import me.prettyprint.cassandra.serializers.IntegerSerializer;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.cassandra.service.ThriftKsDef;
import me.prettyprint.cassandra.service.template.ColumnFamilyTemplate;
import me.prettyprint.cassandra.service.template.ThriftColumnFamilyTemplate;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.factory.HFactory;

public class Template {
	
	private static Cluster cluster =  null;
	@SuppressWarnings({ "rawtypes" })
	private static ColumnFamilyTemplate[] templates = null;
	private static Keyspace ksp = null;
	
	static {
		CassandraHostConfigurator configurator = new CassandraHostConfigurator("127.0.0.1:9160");
		configurator.setRetryDownedHosts(true);
		configurator.setRetryDownedHostsQueueSize(10);
		configurator.setRetryDownedHostsDelayInSeconds(20);
		configurator.setAutoDiscoverHosts(true);
		configurator.setAutoDiscoveryDelayInSeconds(10);
		cluster = HFactory.getOrCreateCluster("my cluster", configurator);
	}
	
	private Template() {
		
	}
	
	public static Keyspace getKeyspace() {
		if (ksp == null) {
			return initCluster();
		}
		return ksp;
	}
	
	@SuppressWarnings("rawtypes")
	public static ColumnFamilyTemplate getTemplate(String columnFamilyName) {
		if (ksp == null) {
			ksp = initCluster();
		}
		
		if (templates == null) {
			templates = new ColumnFamilyTemplate[2];
			templates[0] = new ThriftColumnFamilyTemplate<String, Integer>(ksp,
				       "User", 
                       StringSerializer.get(),        
                       IntegerSerializer.get());
			templates[1] = new ThriftColumnFamilyTemplate<String, String>(ksp,
					"Company",
					StringSerializer.get(),
					StringSerializer.get());
		}
		
		if ("User".equalsIgnoreCase(columnFamilyName)) {
			return templates[0];
		} else if ("Company".equalsIgnoreCase(columnFamilyName)) {
			return templates[1];
		} else {
			return null;
		}
	}
	
	public static Keyspace initCluster() {
		KeyspaceDefinition keyspaceDef = cluster.describeKeyspace("Keyspace_2");
		
		if (keyspaceDef == null) {
		    createSchema();
		    System.out.println("Just created");
		} else {
			System.out.println("Already created");
		}
		
		return HFactory.createKeyspace("Keyspace_1", cluster);
	}
	
	public static void createSchema() {
		// Create User column family definition
		ColumnFamilyDefinition userColumnFamilyDef = HFactory.createColumnFamilyDefinition("Keyspace_2",                              
                "User", 
                ComparatorType.UTF8TYPE);
		
		// Create Company column family definition
		ColumnFamilyDefinition companyColumnFamilyDef = HFactory.createColumnFamilyDefinition("Keyspace_2", 
				"Company", 
				ComparatorType.INTEGERTYPE);
		
		KeyspaceDefinition newKeyspace = HFactory.createKeyspaceDefinition("Keyspace_2",                 
                ThriftKsDef.DEF_STRATEGY_CLASS,  
                1, 
                Arrays.asList(userColumnFamilyDef, companyColumnFamilyDef));
		cluster.addKeyspace(newKeyspace, true);
	}
}
